

package org.apache.ocean;

import java.util.concurrent.*;
import java.util.*;
import org.apache.lucene.document.*;
import java.util.logging.*;
import org.apache.commons.lang.*;
import org.apache.ocean.*;
import org.apache.ocean.database.JDBCDatabase;

/**
 *
 * @author jasonr
 */
// TODO: unit test TransactionLogCleaner
// TODO: make TransactionLogCleaner only delete transactions lower than lowest in index for now
public class TransactionLogCleaner implements Callable {
  public static Logger log = Logger.getLogger(TransactionLogCleaner.class.getName());
  SearchIndexShard searchIndexShard;
  
  public TransactionLogCleaner(SearchIndexShard searchIndexShard) {
    this.searchIndexShard = searchIndexShard;
  }
  
  public Object call() throws Exception {
    Set<TransactionID> unused = getUnused();
    List<TransactionID> lowest = getLowest();
    List<TransactionID> deleted = new ArrayList<TransactionID>(unused.size()+lowest.size());
    SQLTransactionDatabase transactionDatabase = searchIndexShard.getTransactionDatabase();
    List<TransactionID> current = new ArrayList<TransactionID>(10);
    for (TransactionID transactionID : deleted) {
      current.add(transactionID);
      if (current.size() == 10) {
        transactionDatabase.deleteTransactions(current);
        current.clear();
      }
    }
    if (current.size() > 0) transactionDatabase.deleteTransactions(current);
    return new Boolean(true);
  }
  
  private List<TransactionID> getLowest() throws Exception {
    TransactionID lowestTransactionID = searchIndexShard.getLowestTransactionIDIndex();
    SQLTransactionDatabase transactionDatabase = searchIndexShard.getTransactionDatabase();
    List<String> columnNames = new ArrayList<String>(1);
    columnNames.add("id");
    List<TransactionID> transactionIDs = new ArrayList<TransactionID>();
    JDBCDatabase.Results results = transactionDatabase.findTransactionRecordLessResults(lowestTransactionID, columnNames);
    try {
      while (results.hasNext()) {
        TransactionRecord transactionRecord = (TransactionRecord)results.get();
        transactionIDs.add(transactionRecord.id);
      }
      return transactionIDs;
    } finally {
      if (results != null) results.close();
    }
  }
  
  private Set<TransactionID> getUnused() throws Exception {
    CLong numDocs = searchIndexShard.getNumDocuments();
    Set<CLong> deleted = new HashSet<CLong>();
    Map<CLong,CLong> docMap = new HashMap<CLong,CLong>((int)numDocs.value);
    Set<TransactionID> toDelete = new HashSet<TransactionID>();
    
    List<String> columnNames = new ArrayList<String>();
    columnNames.add("id");
    columnNames.add("version");
    columnNames.add("documentID");
    JDBCDatabase.Results results = searchIndexShard.getTransactionDatabase().findTransactionRecordsResults(columnNames, "ORDER BY id DESC");
    try {
      while (results.hasNext()) {
        TransactionRecord transactionRecord = (TransactionRecord)results.get();
        if (StringUtils.equals(TransactionRecord.DELETE, transactionRecord.type)) {
          deleted.add(transactionRecord.documentID);
        } else {
          if (deleted.contains(transactionRecord.documentID)) {
            toDelete.add(transactionRecord.id);
          } else {
            CLong version = docMap.get(transactionRecord.documentID);
            if (version == null) {
              docMap.put(transactionRecord.documentID, transactionRecord.version);
            } else {
              if (version.value > transactionRecord.version.value) {
                toDelete.add(transactionRecord.id);
              }
            }
          }
        }
      }
    } finally {
      if (results != null) results.close();
    }
    log.log(Level.INFO, "toDelete: "+toDelete);
    return null;
  }
}
